package com.example.sendmessage.server.core;

import com.example.sendmessage.common.Consumer;
import com.example.sendmessage.common.ConsumerEnv;
import com.example.sendmessage.common.MqException;
import com.example.sendmessage.server.VirtualHost;

import java.io.IOException;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;

public class ConsumerManager {
    // 需要virtualHost来操作数据
    private VirtualHost virtualHost;

    // 阻塞队列用于存放所有队列的名字，方便后续线程池取用
    private BlockingDeque<String> tokenQueue = new LinkedBlockingDeque<>();

    // 扫描线程，用于扫描需要消费消息的队列，并取出数据
    private Thread scannerThread;

    // 线程池用于执行回调
    private ExecutorService workerPool = null;

    public ConsumerManager(VirtualHost virtualHost) {
        this.virtualHost = virtualHost;
        // 开启扫描线程一直扫描
        scannerThread = new Thread(() -> {
            while (true) {
                try {
                    String queueName = tokenQueue.take();
                    // 查看是否存在该队列
                    Queue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[ConsumerManager] 取令牌时发现队列不存在！queueName = " + queueName);
                    }
                    synchronized (queue) {
                        // 消费消息
                        consumeMessage(queue);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        // 设置为后台线程，程序结束线程随着结束
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    private void consumeMessage(Queue queue) {
        // 按照轮询的方式给队列的消费者依次推送消息
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if (luckyDog == null) {
            // 当前还没有订阅者订阅消息
            return;
        }
        // 取出消息
        Message message = virtualHost.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            return;
        }
        // 执行回调逻辑
        workerPool.submit(() ->{
            try {
                // 将消息先放到待确认的集合中，防止消息丢失
                virtualHost.getMemoryDataCenter().insertMessageWaitAck(queue.getName(), message);
                // 执行回调
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(), message.getBody());
                // 判断是否自动删除
                if (luckyDog.isAutoAck()) {
                    // 如果是持久化的消息就删除硬盘中的消息
                    if (message.getDeliverMode() == 2) {
                        virtualHost.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    // 删除内存消息
                    virtualHost.getMemoryDataCenter().deleteMessage(message.getMessageId());
                    // 删除待确认集合中的消息
                    virtualHost.getMemoryDataCenter().deleteMessageWaitAck(queue.getName(), message.getMessageId());
                }
                System.out.println("[Consumer] 消息消费成功！queueName = "+queue.getName()+
                        ",messageId = " + message.getMessageId());
            } catch (IOException | ClassNotFoundException | MqException e) {
                e.printStackTrace();
            }
        });
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        Queue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MqException("[Consumer] 队列不存在！queueName = "+ queueName);
        }
        // 新建消费者
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (queue) {
            queue.addConsumer(consumerEnv);
            // 判断原来的队列中是否有消息，有就全部消费掉
            int queueMessageNum = virtualHost.getMemoryDataCenter().getQueueMessageNum(queueName);
            for (int i = 0; i < queueMessageNum; i++) {
                consumeMessage(queue);
            }
        }

    }

    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }
}
